#include "stdafx.h"
#include "base/thread/MessageLoopMgr.h"
#include "base/thread/MessageLoopProxy.h"
#include "base/thread/MessagePumpForUI.h"
#include "base/thread/MessageLoop.h"

namespace xBase
{
    VOID CxMessageLoop::AddDestructionObserver(IxDestructionObserver * pDestructionObserver)
    {
        CxAutoLock Guard(m_DestructionObserversListLock);

        std::list<IxDestructionObserver*>::const_iterator it = 
            std::find(m_listDestructionObservers.begin(), m_listDestructionObservers.end(), pDestructionObserver);

        if (it != m_listDestructionObservers.end())
            return;

        m_listDestructionObservers.push_back(pDestructionObserver);
    }

    VOID CxMessageLoop::RemoveDestructionObserver(IxDestructionObserver * pDestructionObserver)
    {
        CxAutoLock Guard(m_DestructionObserversListLock);

        std::list<IxDestructionObserver*>::iterator it = 
            std::find(m_listDestructionObservers.begin(), m_listDestructionObservers.end(), pDestructionObserver);

        if (it != m_listDestructionObservers.end())
            m_listDestructionObservers.erase(it);
    }

    CxMessageLoop * CxMessageLoop::Current() 
    {
        CxMessageLoopMgr * pMessageLoopMgr = MessageLoopMgr::Instance();

        if (!pMessageLoopMgr)
            return NULL;

        CxMessageLoop * pMessageLoop = NULL;
        
        if (!(pMessageLoopMgr->GetMessageLoopPointer(GetCurrentThreadId(), (LONG **)&pMessageLoop)))
            return NULL;

        return pMessageLoop;
    }


    CxMessageLoop::CxMessageLoop(LOOP_TYPE eType, const std::string& strThreadName)
        : m_eType(eType)
        , m_strThreadName(strThreadName)
        , m_bNestableTasksAllowed(TRUE)
        , m_pState(NULL)
        , m_nNextSequenceNum(0)
        , m_bQuitAll(FALSE)
    {
        CxMessageLoopMgr* pMessageLoopMgr = MessageLoopMgr::Instance();
        if (!pMessageLoopMgr)
        {
            assert(FALSE);
        }
        else
        {
            if (!pMessageLoopMgr->SetMessageLoopPointer(GetCurrentThreadId(), (LONG *)this))
            {
                assert(FALSE);
            }
        }

        if (NULL == (m_msgLoopProxy = new CxMessageLoopProxy(this)))
        {
            assert(FALSE);
        }

        // create pump
        if (LOOP_TYPE_UI == m_eType)
        {
            m_Pump = new (std::nothrow) CxMessagePumpForUI();
        } 
        else if (LOOP_TYPE_IO == m_eType)
        {
            assert(FALSE);
            //m_Pump = new(std::nothrow) CMessagePumpForIO;
        }
        else 
        {
            assert(FALSE);
            //m_Pump = new(std::nothrow) CMessagePumpDefault();
        }
    }

    CxMessageLoop::~CxMessageLoop(VOID)
    {
        if (m_msgLoopProxy)
            m_msgLoopProxy->WillDestroyCurrentMessageLoop();

        while (true)
        {
            CxAutoLock Guard(m_DestructionObserversListLock);

            if (m_listDestructionObservers.empty())
                break;

            IxDestructionObserver * pCurObserver = *m_listDestructionObservers.begin();
            m_listDestructionObservers.pop_front();

            // the function will erase from m_listDestructionObservers.
            if (pCurObserver)
                pCurObserver->WillDestroyCurrentMessageLoop();
        }

        // Clean up any unprocessed tasks, but take care: deleting a task could
        // result in the addition of more tasks (e.g., via DeleteSoon).  We set a
        // limit on the number of times we will allow a deleted task to generate more
        // tasks.  Normally, we should only pass through this loop once or twice.  If
        // we end up hitting the loop limit, then it is probably due to one task that
        // is being stubborn.  Inspect the queues to see who is left.
        BOOL bDidWork = FALSE;
        int i = 0;
        for ( ; i < 100; ++ i) 
        {
            DeletePendingTasks();
            ReloadWorkQueue();
            // If we end up with empty queues, then break out of the loop.
            bDidWork = DeletePendingTasks();
            if (!bDidWork)
                break;
        }

        assert(i < 100);

        // OK, now make it so that no one can find us.
        CxMessageLoopMgr* pMessageLoopMgr = MessageLoopMgr::Instance();
        if (!pMessageLoopMgr)
        {
            assert(FALSE);
        }
        else
        {
            if (!(pMessageLoopMgr->SetMessageLoopPointer(GetCurrentThreadId(), (LONG *)NULL)))
            {
                assert(FALSE);
            }
        }
    }

    BOOL CxMessageLoop::GetMessageLoopProxy(CScopedRefPtr<CxMessageLoopProxy>& msgLoopProxy)
    {
        if (m_msgLoopProxy)
        {
            msgLoopProxy = m_msgLoopProxy;
            return TRUE;
        }
        return FALSE;
    }

    VOID CxMessageLoop::Run()
    {
        CxAutoRunState SaveState(this);
        RunHandler();
    }

    VOID CxMessageLoop::RunAllPending() 
    {
        CxAutoRunState SaveState(this);
        m_pState->m_bQuitReceived = TRUE;  // Means run until we would otherwise block.

        RunHandler();
    }

    VOID CxMessageLoop::RunHandler() 
    {
        RunInternal();
    }

    VOID CxMessageLoop::RunInternal() 
    {
        if (m_pState->m_pDispatcher && GetType() == LOOP_TYPE_UI)
            static_cast<CxMessagePumpForUI*>(m_Pump.get())->RunWithDispatcher(this, m_pState->m_pDispatcher);
        else
            m_Pump->Run(this);

        if (m_bQuitAll)
        {
            if (m_pState->m_nRunDepth > 1)
                m_Pump->Quit();
            else
                m_bQuitAll = FALSE;
        }
    }

    BOOL CxMessageLoop::ProcessNextDelayedNonNestableTask()
    {
        if (m_pState->m_nRunDepth != 1)
            return FALSE;

        if (m_DeferredNonNestableWorkQueue.empty())
            return FALSE;

        CTask * pTask = m_DeferredNonNestableWorkQueue.front().m_pTask;
        m_DeferredNonNestableWorkQueue.pop();

        RunTask(pTask);

        return TRUE;
    }

    VOID CxMessageLoop::Quit()
    {
        if (m_pState)
        {
            m_pState->m_bQuitReceived = TRUE;
        } 
    }

    VOID CxMessageLoop::QuitNow() 
    {
        if (m_pState)
        {
            m_Pump->Quit();
        }
    }

    VOID CxMessageLoop::QuitAll()
    {
        m_bQuitAll = TRUE;
    }

    VOID CxMessageLoop::PostTask(CTask * pTask)
    {
        PostTask_Helper(pTask, 0, TRUE);
    }

    VOID CxMessageLoop::PostDelayedTask(CTask * pTask, INT64 nDelayMilliSecond)
    {
        PostTask_Helper(pTask, nDelayMilliSecond, TRUE);
    }

    VOID CxMessageLoop::PostNonNestableTask(CTask * pTask)
    {
        PostTask_Helper(pTask, 0, FALSE);
    }

    VOID CxMessageLoop::PostNonNestableDelayedTask(CTask * pTask, INT64 nDelayMilliSecond)
    {
        PostTask_Helper(pTask, nDelayMilliSecond, FALSE);
    }

    // Possibly called on a background thread!
    VOID CxMessageLoop::PostTask_Helper(CTask * pTask, INT64 nDelayMilliSecond, BOOL bNestable)
    {
        PendingTask oPendingTask(pTask, bNestable);

        if (nDelayMilliSecond > 0)
        {
            oPendingTask.m_DelayedRunTime =
                CTime::Now() + CTimeDelta::FromMilliseconds(nDelayMilliSecond);
        }

        // Warning: Don't try to short-circuit, and handle this thread's tasks more
        // directly, as it could starve handling of foreign threads.  Put every task
        // into this queue.

        CScopedRefPtr<CxMessagePump> Pump;

        {
            CxAutoLock locked(m_IncomingQueueLock);

            INT taskNum = static_cast<INT>(m_IncomingQueue.size());
            m_IncomingQueue.push(oPendingTask);

            if (taskNum > 0)
            {
                return;
            }

            Pump = m_Pump;
        }

        // Since the incoming_queue_ may contain a task that destroys this message
        // loop, we cannot exit incoming_queue_lock_ until we are done with |this|.
        // We use a stack-based reference to the message pump so that we can call
        // ScheduleWork outside of incoming_queue_lock_.

        // trigger start work
        if (Pump)
            Pump->ScheduleWork();
    }

    VOID CxMessageLoop::SetNestableTasksAllowed(BOOL bAllowed)
    {
        if (m_bNestableTasksAllowed != bAllowed) 
        {
            m_bNestableTasksAllowed = bAllowed;
            if (!m_bNestableTasksAllowed)
                return;

            // Start the native pump if we are not already pumping.
            m_Pump->ScheduleWork();
        }
    }

    BOOL CxMessageLoop::NestableTasksAllowed() const 
    {
        return m_bNestableTasksAllowed;
    }

    BOOL CxMessageLoop::IsNested()
    {
        return m_pState->m_nRunDepth > 1;
    }

    INT CxMessageLoop::RunDepth()
    {
        return m_pState->m_nRunDepth;
    }

    INT CxMessageLoop::WaitingTaskNumber()
    {
        CxAutoLock Guard(m_IncomingQueueLock);
        return INT(m_IncomingQueue.size());
    }

    VOID CxMessageLoop::RunTask(CTask * pTask) 
    {
        m_bNestableTasksAllowed = FALSE;

        pTask->Run();
        pTask->Destroy();
        pTask = NULL;

        m_bNestableTasksAllowed = TRUE;
    }

    BOOL CxMessageLoop::DeferOrRunPendingTask(const PendingTask& oPendingTask)
    {
        if (oPendingTask.m_bNestable)
        {
            RunTask(oPendingTask.m_pTask);
            return TRUE;
        }

        m_DeferredNonNestableWorkQueue.push(oPendingTask);
        return FALSE;
    }

    VOID CxMessageLoop::AddToDelayedWorkQueue(const PendingTask& oPendingTask)
    {
        PendingTask oNewPendingTask(oPendingTask);
        oNewPendingTask.m_nSequenceNum = ++m_nNextSequenceNum;
        m_DelayedWorkQueue.push(oNewPendingTask);
    }

    VOID CxMessageLoop::ReloadWorkQueue()
    {
        if (!m_WorkQueue.empty())
            return;  // Wait till we *really* need to lock and load.

        // Acquire all we can from the inter-thread queue with one lock acquisition.
        {
            CxAutoLock lock(m_IncomingQueueLock);
            if (m_IncomingQueue.empty())
                return;

            m_IncomingQueue.Swap(&m_WorkQueue);  // Constant time
        }
    }

    BOOL CxMessageLoop::DeletePendingTasks()
    {
        // clear work queue
        BOOL bDidWork = !m_WorkQueue.empty();
        while (!m_WorkQueue.empty())
        {
            PendingTask oPendingTask = m_WorkQueue.front();
            m_WorkQueue.pop();

            if (!oPendingTask.m_DelayedRunTime.IsNull()) 
            {
                // We want to delete delayed tasks in the same order in which they would
                // normally be deleted in case of any funny dependencies between delayed
                // tasks.
                AddToDelayedWorkQueue(oPendingTask);
            } 
            else
            {
                oPendingTask.m_pTask->Destroy();
                oPendingTask.m_pTask = NULL;
            }
        }

        // clear deferred non nestable work queue
        bDidWork |= !m_DeferredNonNestableWorkQueue.empty();
        while (!m_DeferredNonNestableWorkQueue.empty()) 
        {
            // TODO(darin): Delete all tasks once it is safe to do so.
            // Until it is totaly safe, only delete them under Purify and Valgrind.
            CTask * pTask = NULL;
            pTask = m_DeferredNonNestableWorkQueue.front().m_pTask;
            m_DeferredNonNestableWorkQueue.pop();
            if (pTask)
            {
                pTask->Destroy();
                pTask = NULL;
            }
        }

        // clear delayed work queue
        bDidWork |= !m_DelayedWorkQueue.empty();
        while (!m_DelayedWorkQueue.empty())
        {
            CTask * pTask = m_DelayedWorkQueue.top().m_pTask;
            m_DelayedWorkQueue.pop();
            if (pTask)
            {
                pTask->Destroy();
                pTask = NULL;
            }
        }

        return bDidWork;
    }

    BOOL CxMessageLoop::DoWork() 
    {
        if (!m_bNestableTasksAllowed) 
        {
            // Task can't be executed right now.
            return FALSE;
        }

        for (;;)
        {
            ReloadWorkQueue();
            if (m_WorkQueue.empty())
                break;

            // Execute oldest task.
            do
            {
                PendingTask oPendingTask = m_WorkQueue.front();
                m_WorkQueue.pop();

                if (!oPendingTask.m_DelayedRunTime.IsNull())
                {
                    AddToDelayedWorkQueue(oPendingTask);

                    if (m_DelayedWorkQueue.top().m_pTask == oPendingTask.m_pTask)
                        m_Pump->ScheduleDelayedWork(oPendingTask.m_DelayedRunTime);
                } 
                else 
                {
                    // if one task run, return TRUE
                    if (DeferOrRunPendingTask(oPendingTask))
                        return TRUE;
                }
            } while (!m_WorkQueue.empty());
        }

        return FALSE;
    }

    BOOL CxMessageLoop::DoDelayedWork(CTime * pNextDelayedWorkTime)
    {
        if (!m_bNestableTasksAllowed || m_DelayedWorkQueue.empty())
        {
            *pNextDelayedWorkTime = CTime();
            return FALSE;
        }

        if (m_DelayedWorkQueue.top().m_DelayedRunTime > CTime::Now())
        {
            *pNextDelayedWorkTime = m_DelayedWorkQueue.top().m_DelayedRunTime;
            return FALSE;
        }

        PendingTask oPendingTask = m_DelayedWorkQueue.top();
        m_DelayedWorkQueue.pop();

        if (!m_DelayedWorkQueue.empty())
            *pNextDelayedWorkTime = m_DelayedWorkQueue.top().m_DelayedRunTime;

        return DeferOrRunPendingTask(oPendingTask);
    }

    BOOL CxMessageLoop::DoIdleWork() 
    {
        std::list<IxIdleHandler*>::iterator it = m_listIdleHandler.begin();
        for(; it != m_listIdleHandler.end(); ++it)
        {
            if(*it != NULL)
                (*it)->OnIdle();
        }

        if (ProcessNextDelayedNonNestableTask())
            return TRUE;

        if (m_pState->m_bQuitReceived)
            m_Pump->Quit();

        return FALSE;
    }

    BOOL CxMessageLoop::AddIdleHandler(IxIdleHandler* pHandler)
    {
        if(!pHandler)
            return FALSE;

        std::list<IxIdleHandler*>::iterator it = m_listIdleHandler.begin();
        for(; it != m_listIdleHandler.end(); ++it)
        {
            if (pHandler == *it)
            {
                assert(!"dumplicate add IxIdleHandler");
                return FALSE;
            }
        }

        m_listIdleHandler.push_back(pHandler);
        return TRUE;

    }

    BOOL CxMessageLoop::RemoveIdleHandler(IxIdleHandler* pHandler)
    {
        if(!pHandler)
            return FALSE;

        std::list<IxIdleHandler*>::iterator it = m_listIdleHandler.begin();
        for(; it != m_listIdleHandler.end(); ++it)
        {
            if (pHandler == *it)
            {
                m_listIdleHandler.erase(it);
                return TRUE;
            }
        }
        assert(L"Haven't find IxIdleHandler");
        return FALSE;
    }

    VOID CxMessageLoop::DoIdleHandler()
    {
        std::list<IxIdleHandler*>::iterator it = m_listIdleHandler.begin();

        for(; it != m_listIdleHandler.end(); ++it)
        {
            if(*it != NULL)
                (*it)->OnIdle();
        }
    }

    // store the run state of pMessageLoop
    CxMessageLoop::CxAutoRunState::CxAutoRunState(CxMessageLoop * pMessageLoop)
        : m_pMessageLoop(pMessageLoop) 
    {
        m_pPreviousState = pMessageLoop->m_pState;
        if (m_pPreviousState)
        {
            m_nRunDepth = m_pPreviousState->m_nRunDepth + 1;
        }
        else 
        {
            m_nRunDepth = 1;
        }
        pMessageLoop->m_pState = this;

        // Initialize the other fields:
        m_bQuitReceived = false;

        m_pDispatcher = NULL;
    }

    CxMessageLoop::CxAutoRunState::~CxAutoRunState()
    {
        m_pMessageLoop->m_pState = m_pPreviousState;
    }

    bool CxMessageLoop::PendingTask::operator< (const PendingTask& Other) const 
    {
        if (m_DelayedRunTime < Other.m_DelayedRunTime)
            return false;

        if (m_DelayedRunTime > Other.m_DelayedRunTime)
            return true;

        // If the times happen to match, then we use the sequence number to decide.
        // Compare the difference to support integer roll-over.
        return (m_nSequenceNum - Other.m_nSequenceNum) > 0;
    }
}